package com.health.trans.channel;


import cn.hutool.core.collection.CollUtil;
import com.health.es.entity.log.TransformLog;
import com.health.trans.common.util.GenericUtil;
import com.health.trans.util.CacheMap;
import lombok.Getter;
import lombok.Setter;
import lombok.SneakyThrows;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.function.Consumer;

/**
 * 内存通道
 * 内存Channel的具体实现，底层其实是一个ArrayBlockingQueue
 *
 * @author nqlz
 * @date 2022/04/24
 */
@Slf4j
@Setter
@Getter
@Accessors(chain = true)
public class MemorySet {

    private Channel<TransformLog> transformLogSet;
    private CacheMap<Class<?>, Channel<?>> mySqlSet = new CacheMap<>();
    protected int capacity = 1000;
    protected Consumer<?> lineHandler;

    public MemorySet() {
        transformLogSet = new Channel<>(capacity, capacity);
    }

    public MemorySet(int capacity) {
        transformLogSet = new Channel<>(capacity, capacity);
        this.capacity = capacity;
    }

    public void pushLog(TransformLog r) {
        transformLogSet.push(r);
    }

    public <T> Collection<T> checkMysqlQueue(Class<T> tClass) {
        Channel<T> metaChannel = (Channel<T>) mySqlSet.get(tClass);
        if (null != metaChannel && metaChannel.getQueue().remainingCapacity() < 500) {
            ArrayList<T> c = new ArrayList<>();
            metaChannel.pullAll(c);
            return c;
        }
        return Collections.emptyList();
    }

    public Collection<TransformLog> checkLogQueue() {
        if (transformLogSet.getQueue().remainingCapacity() < 500) {
            ArrayList<TransformLog> c = new ArrayList<>();
            transformLogSet.pullAll(c);
            return c;
        }
        return Collections.emptyList();
    }


    public void pushLogAll(Collection<TransformLog> rs) {
        transformLogSet.pushAll(rs);
    }

    public <T> void pushOther(Class<T> tClass, T r) {
        lineHandler.accept(GenericUtil.get(checkMysqlQueue(tClass)));
        Channel<T> queue = GenericUtil.get(mySqlSet.get(tClass, () -> {
            Channel<T> tMetaChannel = new Channel<>(capacity, capacity);
            mySqlSet.put(tClass, tMetaChannel);
            return tMetaChannel;
        }));
        queue.push(r);
    }


    public void clear() {
        transformLogSet.clear();
        if (CollUtil.isNotEmpty(mySqlSet)) {
            mySqlSet.forEach((k, v) -> v.clear());
        }
    }

}
